package hylanda.processors.gethead;

import java.io.BufferedOutputStream;
import java.io.OutputStream;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import org.apache.commons.io.IOUtils;
import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.flowfile.attributes.CoreAttributes;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.util.StandardValidators;

import com.alibaba.fastjson.JSONObject;

import hylanda.processors.util.GenUrlCrc64;
import hylanda.processors.util.HtmlHeadUtil;
import hylanda.processors.util.NifiUtil;

@Tags({ "http head code Content-Length" })
@CapabilityDescription("Provide a description")
@SeeAlso({})
@ReadsAttributes({ @ReadsAttribute(attribute = "", description = "") })
@WritesAttributes({ @WritesAttribute(attribute = "", description = "") })
public class GetHeadProcessor extends AbstractProcessor {

	public static final PropertyDescriptor URL_FIELD = new PropertyDescriptor.Builder().name("URL_FIELD")
			.displayName("url field").description("URL filed,默认是url")
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor IS_PROXY = new PropertyDescriptor.Builder().name("isproxy")
			.displayName("isproxy").description("isproxy").allowableValues(new String[] { "true", "false" })
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor PROXY_IP = new PropertyDescriptor.Builder().name("proxyip")
			.displayName("proxyip").description("proxyip").addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor PROXY_PORT = new PropertyDescriptor.Builder().name("proxyport")
			.displayName("proxyport").description("proxyport").addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
			.build();
	public static final PropertyDescriptor CONNECT_TIMEOUT = new PropertyDescriptor.Builder().name("connectTimeOut")
			.displayName("connectTimeOut").description("connect http timeout")
			.addValidator(StandardValidators.NON_EMPTY_VALIDATOR).build();
	public static final PropertyDescriptor READ_TIMEOUT = new PropertyDescriptor.Builder().name("readTimeOut")
			.displayName("readTimeOut").description("readtimeout").addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
			.build();

	public static final Relationship SUCCESS = new Relationship.Builder().name("SUCCESS").description("获取到状态值").build();
	public static final Relationship FAILURE = new Relationship.Builder().name("FAILUER").description("未获取到状态值")
			.build();

	private List<PropertyDescriptor> descriptors;

	private Set<Relationship> relationships;

	@Override
	protected void init(final ProcessorInitializationContext context) {
		final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
		descriptors.add(IS_PROXY);
		descriptors.add(PROXY_IP);
		descriptors.add(PROXY_PORT);
		descriptors.add(CONNECT_TIMEOUT);
		descriptors.add(READ_TIMEOUT);
		descriptors.add(URL_FIELD);
		this.descriptors = Collections.unmodifiableList(descriptors);

		final Set<Relationship> relationships = new HashSet<Relationship>();
		relationships.add(SUCCESS);
		relationships.add(FAILURE);
		this.relationships = Collections.unmodifiableSet(relationships);
	}

	@Override
	public Set<Relationship> getRelationships() {
		return this.relationships;
	}

	@Override
	public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
		return descriptors;
	}

	@OnScheduled
	public void onScheduled(final ProcessContext context) {

	}

	@SuppressWarnings("deprecation")
	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		FlowFile flowFile = session.get();
		if (flowFile == null) {
			return;
		}
		String content = NifiUtil.getContent(session, flowFile);
		// ComponentLog log = getLogger();
		String urlField = null;
		try {
			urlField = context.getProperty(URL_FIELD).getValue();
		} catch (Exception e) {
			// TODO: handle exception
		}
		String isagent = null;
		try {
			isagent = context.getProperty(IS_PROXY).getValue();
		} catch (Exception e) {
			// TODO: handle exception
		}
		String porxyIp = null;
		try {
			porxyIp = context.getProperty(PROXY_IP).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}
		String porxyPort = null;
		try {
			porxyPort = context.getProperty(PROXY_PORT).getValue().trim();
		} catch (Exception e) {
			// TODO: handle exception
		}

		String connectTimeOut = null;
		try {
			connectTimeOut = context.getProperty(CONNECT_TIMEOUT).getValue();// flowFile.getAttribute("connectTimeOut");
		} catch (Exception e) {
			// TODO: handle exception
		}

		String readTimeOut = null;
		try {
			readTimeOut = context.getProperty(READ_TIMEOUT).getValue();// flowFile.getAttribute("readTimeOut");
		} catch (Exception e) {
			// TODO: handle exception
		}

		if (null == urlField || urlField.equals("")) {
			urlField = "url";
		}
		if (null == content || "".equals(content)) {
			flowFile = session.write(flowFile, (in, out) -> {
				try (OutputStream outputStream = new BufferedOutputStream(out)) {
					outputStream.write(IOUtils.toByteArray("content为空"));
				}
			});
			session.transfer(flowFile, FAILURE);
			return;
		}
		//JSONObject contentJson = new JSONObject();
		try {
			JSONObject contentJson = JSONObject.parseObject(content);
			
			if (null == contentJson || "".equals(contentJson)) {
				flowFile = session.write(flowFile, (in, out) -> {
					try (OutputStream outputStream = new BufferedOutputStream(out)) {
						outputStream.write(IOUtils.toByteArray("content转换json失败"));
					}
				});
				session.transfer(flowFile, FAILURE);
				return;
			}
			String url = contentJson.getString(urlField);
			HtmlHeadUtil htmlHeadUtil = new HtmlHeadUtil();
			StringBuilder sb = new StringBuilder();
			// log.info("GetHeadProcessor url = "+url);
			long startTime = System.currentTimeMillis();
			Map<String, String> map = htmlHeadUtil.getHeadersNew(url, sb, isagent, porxyIp, porxyPort, connectTimeOut,
					readTimeOut);
			long endTime = System.currentTimeMillis();
			sb.append("head " + (endTime - startTime) + "ms");
			if (map == null || map.equals("")) {
				flowFile = session.write(flowFile, (in, out) -> {
					try (OutputStream outputStream = new BufferedOutputStream(out)) {
						outputStream.write(IOUtils.toByteArray(sb.toString()));
					}
				});
				session.transfer(flowFile, FAILURE);
				return;
			}
			JSONObject resultJson = JSONObject.parseObject(content);
			String code = map.get("code");
			String contentLength = map.get("Content-Length");
			if (map.containsKey("timeout")) {
				resultJson.put("code", "timeout");
			}
			
			resultJson.put("code", code);
			resultJson.put("Content-Length", contentLength);
			if (map.containsKey("newUrl")) {
				resultJson.put("newUrl", map.get("newUrl"));
			}

			flowFile = session.write(flowFile, (in, out) -> {
				try (OutputStream outputStream = new BufferedOutputStream(out)) {
					outputStream.write(IOUtils.toByteArray(resultJson.toJSONString()));
				}
			});

			String uuid = GenUrlCrc64.GenCrc64(UUID.randomUUID().toString());
			String session_id = flowFile.getAttribute(CoreAttributes.UUID.key());
			String instance_id = getIdentifier();
			JSONObject logJSON = new JSONObject();
			JSONObject msgJSON = new JSONObject();

			Date date = new Date();
			SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			String time = sdf.format(date);
			msgJSON.put("time", time);
			msgJSON.put("status", "success");

			logJSON.put("msg", msgJSON);
			logJSON.put("session_id", session_id);
			logJSON.put("uuid", uuid);
			logJSON.put("instance_id", instance_id);
			getLogger().billing(session_id);
			getLogger().info(logJSON.toJSONString());
			session.transfer(flowFile, SUCCESS);
			
			
		} catch (Exception e) {
			flowFile = session.write(flowFile, (in, out) -> {
				try (OutputStream outputStream = new BufferedOutputStream(out)) {
					outputStream.write(IOUtils.toByteArray("content转换json失败"));
				}
			});
			session.transfer(flowFile, FAILURE);
			return;
		}

		
	}

}
